package com.wyy.test.akkaimpl;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @Date 2024/7/30
 * @Author wyy
 **/
public abstract class AbstractActor {

    private final BlockingQueue<Object> mailBox = new LinkedBlockingQueue<>();
    private final Executor executor;
    private boolean handleMsgStarted = false;

    public abstract void onReceive(Object message);

    public boolean tell(Object message) {
        System.out.println("receive msg: " + message);
        boolean offered = mailBox.offer(message);
        if (!handleMsgStarted) {
            handleMsgStarted = true;
            System.out.println("start handle msg thread");
            new Thread(this::startHandleMsg).start();
        }
        return offered;
    }

    public AbstractActor(Executor executor) {
        this.executor = executor;
    }

    private void startHandleMsg() {
        AtomicReference<Boolean> handleMsg = new AtomicReference<>(true);
        while (!mailBox.isEmpty()) {
            if (!handleMsg.get()) {
                continue;
            }
            Object msg = null;
            try {
                msg = mailBox.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
                break;
            }
            System.out.println("get msg from mailbox: " + msg);
            final Object finalMsg = msg;
            handleMsg.set(false);
            executor.execute(() -> {
                onReceive(finalMsg);
                handleMsg.set(true);
            });
        }
    }
}
